package jmind.core.redis;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import jmind.core.algo.QuickLZ;
import jmind.core.util.SerializeUtil;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPipeline;
import redis.clients.jedis.ShardedJedisPool;

/**
 * sharded  redis 实现
 * @author weibo-xie
 * 2012-11-6
 */
public class ShardedRedis extends AbstractJedis {
    private final ShardedJedisPool shardPool;

    /**
     * 
     * @param redisConfig  格式为10.22.32.143:64000,10.22.32.152:64000,10.22.32.144:64000
     */
    public ShardedRedis(String hosts, int timeout, int maxActive, int maxIdle) {
        JedisPoolConfig config = new JedisPoolConfig();
        // #最大分配的对象数  
        config.setMaxActive(maxActive);
        // 最大能够保持idel状态的对象数
        config.setMaxIdle(maxIdle);
        config.setMaxWait(1000);
        config.setTestOnBorrow(true);
        List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
        for (String host : hosts.split(",")) {
            String[] hostAndPort = host.split(":");
            JedisShardInfo info = new JedisShardInfo(hostAndPort[0], Integer.parseInt(hostAndPort[1]));
            info.setTimeout(timeout * 1000);
            shards.add(info);
        }

        shardPool = new ShardedJedisPool(config, shards);

    }

    @Override
    public ShardedJedis getResource() {
        return shardPool.getResource();
    }

    @Override
    public void close(Object resource) {
        if (resource != null)
            shardPool.returnResourceObject(resource);

    }

    /******************string******************/

    public long del(String... keys) {
        long i = 0;
        ShardedJedis jedis = getResource();
        try {
            for (String key : keys) {
                i += jedis.del(key);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            close(jedis);
        }
        return i;
    }

    public List<String> mget(String... keys) {
        ShardedJedis jedis = getResource();
        List<String> list = new ArrayList<String>();
        try {
            for (String key : keys) {
                list.add(jedis.get(key));
            }
        } finally {
            close(jedis);
        }
        return list;
    }

    public Long zaddKV(String key, Map<String, Double> memberScores) {
        ShardedJedis jedis = getResource();
        try {
            ShardedJedisPipeline pipelined = jedis.pipelined();
            for (Entry<String, Double> entry : memberScores.entrySet()) {
                pipelined.zadd(key, entry.getValue(), entry.getKey());
            }
            pipelined.sync();
        } finally {
            close(jedis);
        }
        return 1l;
    }

    @Override
    public void releaseResource() {
        shardPool.destroy();

    }

    public boolean setObject(String key, Object o) {
        ShardedJedis jedis = getResource();
        try {
            // 序列化
            byte[] byteArray = SerializeUtil.serialize(o);
            // 压缩
            byteArray = QuickLZ.compress(byteArray, 3);
            String setObjectRet = jedis.set(key.getBytes(), byteArray);
            return "OK".equalsIgnoreCase(setObjectRet);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            close(jedis);
        }
        return false;
    }

    @SuppressWarnings("unchecked")
    public <T> T getObject(String key, Class<T> clazz) {
        ShardedJedis jedis = getResource();
        Object o = null;
        try {
            byte[] bs = jedis.get(key.getBytes());
            // 解压
            bs = QuickLZ.decompress(bs);
            o = SerializeUtil.unserialize(bs);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            close(jedis);
        }
        return (T) o;
    }

}
